#include <UTILS/SBuffer.h>
#include <PECAgent/PEContext.h>
#include <DPS/PH/DpsFunctions.h>
#include <DPS/PH/DpsOPContext.h>
#include "UDOP_OpGlobalY.h"
#include "analysis.h"
#include "utilities.h"

SPC_NAMESPACE_USE;
UTILS_NAMESPACE_USE;
using namespace DPS;
using namespace std;
using namespace estd;

/*
 * Function used to initialize the operator
 */
void UDOP_OpGlobalY::initializeOperator() 
{

  char str10[150];
  const DpsOPContext &context = this->getContext();
  const std::string &args = context.getCmdArguments();
  strcpy(str10, args.c_str());
  sscanf(str10,"%*s %*c%d%*c %*s %*c%d%*c",&num_ensemble,&astep);
  
  tmpcon = (double **)malloc(num_ensemble*sizeof(double *) );
  tmplat = (float **)malloc(num_ensemble*sizeof(float *) );
  tmplon = (float **)malloc(num_ensemble*sizeof(float *) );
  tmpidx = (int **)malloc(num_ensemble*sizeof(int *) );

  en_numline = (int *)malloc(num_ensemble*sizeof(int));

  int i = 0;
  for(i=0; i<num_ensemble; i++ )
  {
	  tmpcon[i] = NULL;
	  tmplat[i] = NULL;
	  tmplon[i] = NULL;
	  tmpidx[i] = NULL;
  }
  // Create a user defined metric
  // opm.createMetric("intuples","Number of ingested tuples");
}

/*
 * Function used to finalize the operator
 */
void UDOP_OpGlobalY::finalizeOperator() 
{

// free space
	int i= 0;
	for (i=0; i<num_ensemble; i++ )
	{
	if (tmpcon[i]!=NULL)
		free(tmpcon[i]);
	
	if (tmplat[i]!=NULL)
		free(tmplat[i]);
	
	if (tmplon[i]!=NULL)
		free(tmplon[i]);
	
	if (tmpidx[i]!=NULL)
		free(tmpidx[i]);
	}
	free(tmpcon);
	free(tmplat);
	free(tmplon);
	free(tmpidx);
}

/*
 * Function used to process command line arguments
 */
void UDOP_OpGlobalY::processCmdArgs(const string & /*args*/) 
{
  
  // ArgContainer arg(args); // include "ArgContainer.h"
  // arg.argc and arg.argv are now available for use
  // do other initialization here
}


/*
 * Process function for input port 0
 */
void UDOP_OpGlobalY::processInput0(const IPort0_t & tuple)
{
  SPCDBG(L_DEBUG, "Processing tuple from input port 0 " << tuple, "dpsop");
  int i = 0;
  int stid = 0; // the streamid
  const int &sid = (const int&) (tuple.get_sid());
  const dlist<double> &con = (const dlist<double>&) (tuple.get_con());
  const dlist<float> &lat = (const dlist<float>&) (tuple.get_lat());
  const dlist<float> &lon = (const dlist<float>&) (tuple.get_lon());
  int len = con.size();
  bool flag = tuple.get_flag();
  stid = sid;
  en_numline[stid] = len;

  if ((len > 0) && flag )
  {
  // Allocate space for each row in the tmp* array.
    tmpcon[stid] = (double *)realloc(tmpcon[stid], len*sizeof(double) );
    tmplat[stid] = (float *)realloc(tmplat[stid], len*sizeof(float) );
    tmplon[stid] = (float *)realloc(tmplon[stid], len*sizeof(float) );
	tmpidx[stid] = (int *)realloc(tmpidx[stid], len*sizeof(int) );

  // Store the incoming tuples
    for (i=0; i<len; i++)
    {
    tmpcon[stid][i] = con[i];
    tmplat[stid][i] = lat[i];
    tmplon[stid][i] = lon[i];
    }
	
	gflag = true;
  }
  else
	  gflag = false;

  cnt++;// Accumulate tuples and handle at once
		// simulate the tubmling window
  if(cnt%num_ensemble == 0 )
  {
	  TransOpEn();
	  cnt = 0;
  }
}

/*
 * Punctuate function
 */

void UDOP_OpGlobalY::TransOpEn()
{
	int i, j, k, l;
	int ix, iy;
	float x,y,sx,sy,px,py;
	int length2d = 0;
	int counter = 0;

	OPort0_t otuple;
	dlist<double> &yb_bar = otuple.get_yb_bar();
    dlist<float> &yb_clat = otuple.get_yb_clat();
	dlist<float> &yb_clon = otuple.get_yb_clon();
	
	if (gflag)
	{	
	read_control(&x,&y,&sx,&sy,&px,&py);
	length2d = (int)(px/sx+1)*(int)(py/sy+1);
	int length1d = (int)(px/sx+1);
	dlist<int16_t> yb_idx; 	
	//1. Calculate idx for each grid with concentration.
	for (i=0; i<num_ensemble; i++)
		for (j=0; j<en_numline[i]; j++)
		{
			ix = (int)(floor((tmplat[i][j]-(x-(sx+px)/2.0))/sx));
			iy = (int)(floor((tmplon[i][j]-(y-(sy+py)/2.0))/sy));
			tmpidx[i][j] = iy * length1d + ix;
		}
	
		
	//2. Allocate space for array1 array2 of length x*y
	double *array1;
	int *array2;
	float *array3;
	float *array4;
	array1 = (double *)malloc(length2d*sizeof(double));
	array2 = (int *)malloc(length2d*sizeof(int));
	array3 = (float *)malloc(length2d*sizeof(float));
	array4 = (float *)malloc(length2d*sizeof(float));
	for (i=0; i<length2d; i++)
	{
		array1[i] = 0.0;
		array2[i] = 0;
		array3[i] = 0.0;
		array4[i] = 0.0;
	}

	for (i=0; i<num_ensemble; i++)
		for (j=0; j<en_numline[i]; j++)   // check?
			{
			array1[tmpidx[i][j]] += tmpcon[i][j];
			array2[tmpidx[i][j]]++;
			array3[tmpidx[i][j]] += tmplat[i][j];
			array4[tmpidx[i][j]] += tmplon[i][j];
			}
	// Find average of Yb_g: yb_bar_g_o
	counter = 0;
	for (i=0; i<length2d; i++)
		{
		if (array2[i]!=0)
			counter++;
		}

	yb_idx.reserve(counter);
	//using namespace boost::numeric::ublas;
	dmatrix<double> en_con1 = DPS::makeDoubleMatrix(num_ensemble, counter);
	dmatrix<double> &en_con = en_con1;
	for (i=0; i<num_ensemble; i++ )
		for (j=0; j<counter; j++ )
			en_con(i,j) = 0.0;
	
	k = 0; //reset
	for (i=0; i<length2d; i++)
		{
		if (array2[i]!=0)
			{
			yb_bar.push_back(array1[i] / (double)array2[i] );
			yb_clat.push_back(array3[i] / (float)array2[i] );
			yb_clon.push_back(array4[i] / (float)array2[i] );
			yb_idx[k] = i; //keep position info
			k++;
			}
		}
	// Allocate space for Yb_g, num_ensemble * cnt;
	for (i=0; i<num_ensemble; i++)
		for (j=0; j<counter; j++)
		{
		for (l=0; l<en_numline[i]; l++)
			{
			if (yb_idx[j] == tmpidx[i][l])
				{
				en_con(i,j) += (tmpcon[i][l] - yb_bar[j]);

				}
			}
		}
/*	double sumt = 0.0;
	for(i=0; i<counter; i++)
	{sumt = 0.0;
		for(j=0; j<num_ensemble; j++)
		{
			sumt+=en_con(j,i);
		}
		cout<<sumt<<endl;
	}
*/
		if (array1!=NULL)	
		free(array1);
		if (array2!=NULL)
		free(array2);
		if (array3!=NULL)
		free(array3);
		if (array4!=NULL)
		free(array4);

	otuple.set_en_con(en_con);
	}

	otuple.set_flag(gflag);

	submit0(otuple);
}
/*
 * Function used to notify the operator about the readiness of all ports (this
 * is useful when submitting tuples from within a context other than the process
 * functions of the operator).
 */ 
/*
void UDOP_OpGlobalY::allPortsReady() 
{
} 
*/

/*
 * Terminate function, called asynchronously
 */
/*
void UDOP_OpGlobalY::prepareToTerminate() 
{
}
*/

/*
 * Function used to update performance counters
 */
/*
static void updatePerformanceCounters(DpsOPMetrics & opm) 
{ 
  static long long ntuples = 0;
  // This is not the ideal way of using performance counters
  // Ideally one should store the result from getMetricByName, 
  // in a class member variable. For instance:
  //   DpsMetric & m = opm.getMetricByName("intuples");
  // m could be made a member variable of UDOP_OpGlobalY
  // Then, the following is a more efficient way to use the counter:
  //   m.setValue(ntuples++);
  opm.getMetricByName("intuples").setValue(ntuples++);
}
*/

/*
 * Checkpointing function for saving state, called asynchronously 
 */
/*
void UDOP_OpGlobalY::getCheckpoint(SBuffer & checkpoint) 
{
  SPCDBG(L_DEBUG, "Getting operator state...", "dpsop");
}
*/

/*
 * Checkpointing function for restoring state, called asynchronously
 */
/*
void UDOP_OpGlobalY::restoreCheckpoint(SBuffer & checkpoint) 
{
  SPCDBG(L_DEBUG, "Restoring operator state...", "dpsop");
}
*/


